package com.errol.batchprocessor.batchprocessor.advisorimpl.batchsplitter;

import com.renjia.entity.AbstractImportBatch;
import com.renjia.service.generic.middleware.batchprocessor.bo.ThreadPoolStatusBO;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractBatchPayload;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractPiecePayload;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractSinglePayload;
import org.springframework.lang.NonNull;

import java.util.Collections;
import java.util.List;

/**
 * 按批次大小分片
 *
 * @author zhang xiao bin
 * @version v1.0
 * @date 2021/03/12
 * @since v5.0.6
 **/
public class PieceSizeBatchSplitter<IE extends AbstractImportBatch, AT extends Enum<?>, BE, SE, BP extends AbstractBatchPayload<IE, AT, BE, SE>, PP extends AbstractPiecePayload<SE, SP>, SP extends AbstractSinglePayload<SE>>
        implements BatchSplitter<IE, AT, BE, SE, BP, PP, SP> {

    /**
     * 分片最大长度，限定 [100,500] 之间
     */
    private final float maximumSizeOfPiece;

    public PieceSizeBatchSplitter(int maximumSizeOfPiece) {
        maximumSizeOfPiece = Math.min(maximumSizeOfPiece, 500);
        maximumSizeOfPiece = Math.max(maximumSizeOfPiece, 100);
        this.maximumSizeOfPiece = maximumSizeOfPiece;
    }

    /**
     * 根据批次记录数量，按下列判断依次执行：
     * - 100以内，单线程
     * - 1000以内根据批次大小平分，保证每个分片的长度在 [50, 100] 之间
     * - 1000以上进行平分，每个分片长度在 (maximumSizeOfPiece*2/3, maximumSizeOfPiece] 之间
     *
     * @param batchPayload        批次数据
     * @param threadPoolStatus    线程池状态
     * @param batchSplitterHelper 批次添加到批处理器的扩展点
     * @return PP 列表
     */
    @Override
    public List<PP> doSplit(@NonNull BP batchPayload, @NonNull ThreadPoolStatusBO threadPoolStatus, @NonNull BatchSplitterHelper<IE, AT, BE, SE, BP, PP, SP> batchSplitterHelper) {
        int size = batchPayload.getSingleEntityList().size();
        if (size <= 100) {
            // 单线程
            return Collections.singletonList(batchSplitterHelper.newPiecePayload(batchPayload.getSingleEntityList()));
        }
        int threads;
        if (size <= 1000) {
            // 按记录条数，最少 2 个线程，最多 10 个线程；每个线程最少 50 条，最多 100 条
            threads = (int) Math.ceil(size / 100.0);
        } else {
            threads = (int) Math.ceil(size / maximumSizeOfPiece);
        }
        return doSplitByPieceCount(size, threads, batchPayload, batchSplitterHelper);
    }
}
